package com.zuora.interview.topn.processor.step2;

import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.zuora.interview.topn.bean.Constants;
import com.zuora.interview.topn.util.Clock;
import com.zuora.interview.topn.util.LogHelper;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Created by chunyang.gao on 2016/8/23.
 */
public class UserPartitionerProcessor {

    private static final Logger logger = LoggerFactory.getLogger(UserPartitionerProcessor.class);

    private static final Logger console = LoggerFactory.getLogger("console");

    Splitter commaSplitter = Splitter.on(",");

    public void work() throws IOException {
        Clock clock = Clock.start();
       File d = new File(Constants.logPath+"step1");
        final CountDownLatch listener = new CountDownLatch(d.listFiles().length);
        for(final File f:d.listFiles()){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        new UserPartitionerProcessor().test(f);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }finally {
                        listener.countDown();
                    }
                }
            }).start();
        }
        try {
            listener.await(30, TimeUnit.SECONDS);
            console.warn("step2 in {} ms", clock.checkTime());
        } catch (InterruptedException e) {
            console.warn("step2 in " + clock.checkTime() + " ms as error", e);
        }
    }

    private void test(File f) throws IOException {
        LineIterator it = FileUtils.lineIterator(f);
        Map<String,List<String>> accessByUser = Maps.newHashMap();
        try {
            while (it.hasNext()) {
                String line = it.nextLine();
                List<String> parts = commaSplitter.splitToList(line);
                String username = parts.get(0);
                String path = parts.get(1);
                List<String> accessRecords = accessByUser.get(username);
                if(CollectionUtils.isEmpty(accessRecords)){
                    accessRecords = Lists.newArrayList();
                    accessByUser.put(username,accessRecords);
                }
                accessRecords.add(path);
            }
        } finally {
            LineIterator.closeQuietly(it);
        }
        for(String username:accessByUser.keySet()){
            List<String> records = accessByUser.get(username);
            if(CollectionUtils.isEmpty(records) || records.size()<3){
                continue;
            }

            for(int i=0;i<=records.size()-3;i++){
                String curr3PP = records.get(i)+"-"+records.get(i+1)+"-"+records.get(i+2);
                int partionId = generateHash(curr3PP);
                LogHelper.set(partionId + "");
                logger.info(curr3PP+","+username);
                LogHelper.clear();
            }
        }

    }

    private int generateHash(String username){
        char first = 0;
        if(username.length()>1){
            first = username.charAt(username.length()-2);
        }
        char second = username.charAt(username.length()-1);
        String joinedChar = (first+0)+""+(second+0);
        return Integer.valueOf(joinedChar)%1024;
    }
}
